package com.zghhost.codedemo.pulsar;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;

/**
 * @author guohua.zhang@zhaopin.com.cn
 * @Date 2019/6/19 14:44
 */
public class PulsarConsumer {
    public static void main(String[] args)throws Exception{
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://172.17.5.175:6650")
                .build();

        sub(client,"persistent://platform/dp/job.solr.result","job更新时通知solr");

        sub(client,"persistent://platform/dp/job.cache.clear","清空caption缓存");

        sub(client,"persistent://platform/dp/job.solr.delete","job删除时通知solr");
    }

    public static void sub(PulsarClient client,String topic,String desc)throws Exception{
        new Thread(){
            @Override
            public void run() {
                try{
                    Consumer consumer = client.newConsumer()
                            .topic(topic)
                            .subscriptionName(topic)
                            .subscribe();

                    while (true){
                        Message msg = consumer.receive();
                        try {
                            System.out.printf("%s: %s \n",desc, new String(msg.getData()));
                            consumer.acknowledge(msg);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }.start();
    }
}
